package com.ibm.cps.sql;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import com.google.common.collect.HashMultimap;
import com.ibm.cps.newmessage.BasicDataSourceMetadata;
import com.ibm.cps.sql.expression.BooleanExpression;
import com.ibm.cps.sql.expression.ValueExpression;
import com.ibm.factories.DataSourceFactory;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.ErrorCode;
import com.ibm.util.SchemaCache;
import com.ibm.util.exception.CPSException;

public class ProjectionParser {
	private final String tenantid;
	private final DataSourceFactory datasourceFactory;
	private final SchemaCache schemaCache;

	public ProjectionParser(String tenantid, DataSourceFactory datasourceFactory,
			SchemaCache schemaCache) {
		this.tenantid = tenantid;
		this.datasourceFactory = datasourceFactory;
		this.schemaCache = schemaCache;
	}

	public Set<DatasourceField> parseProjection(List<String> dataSources, SelectItemDescriptor selectItems, AggregationDescriptor aggregation, BooleanExpression filterCondition) throws CPSException{
		FieldUsage usage = new FieldUsage();
		collectFieldUsageFromDataSources(usage, dataSources);
		collectUsageFromSelectItems(usage, dataSources, selectItems);
		collectUsageFromCondition(usage, filterCondition);
		updateAggregations(usage, aggregation);
		return usage.getFieldsForProjection();
	}

	private void collectFieldUsageFromDataSources(FieldUsage usage, List<String> dataSources) throws CPSException{
		for (String datasource : dataSources) {
			BasicDataSourceMetadata metadata = datasourceFactory.getDataSourceMetadata(datasource);
			ITimeSeriesSchema schema = schemaCache.getSchema(tenantid, metadata.getSchemaid(), new ArrayList<String>());
			String[] fieldNames = schema.getTagNames();
			for (String shortName : fieldNames) {
				String fullName = datasource + ":" + shortName;
				if(!usage.isDatasourceField(shortName)){
					usage.addFullNameMapping(shortName, fullName);
					usage.addShortNameMapping(fullName, shortName);
				}
				usage.addFieldBelonging(shortName, datasource);
				usage.addFieldBelonging(fullName, datasource);
			}
		}
	}

	private void collectUsageFromSelectItems(FieldUsage usage, List<String> datasources, SelectItemDescriptor selectItems) throws CPSException{
		Set<String> selectFieldsFullNames = new HashSet<>();
		if(selectItems.isGlobalSelectAll()){
			for (String datasource : datasources) {
				BasicDataSourceMetadata metadata = datasourceFactory.getDataSourceMetadata(datasource);
				ITimeSeriesSchema schema = schemaCache.getSchema(tenantid, metadata.getSchemaid(), new ArrayList<String>());
				String[] fieldNames = schema.getTagNames();
				for (String field : fieldNames) {
					selectFieldsFullNames.add(datasource + ":" + field);
				}
			}
		}
		List<String> allSelectedDatasources = selectItems.getSelectAllTables();
		for (String datasource : allSelectedDatasources) {
			BasicDataSourceMetadata metadata = datasourceFactory.getDataSourceMetadata(datasource);
			ITimeSeriesSchema schema = schemaCache.getSchema(tenantid, metadata.getSchemaid(), new ArrayList<String>());
			String[] fieldNames = schema.getTagNames();
			for (String field : fieldNames) {
				selectFieldsFullNames.add(datasource + ":" + field);
			}
		}

		// Check if there is any non-existence or ambigious field in select item
		Map<String, ValueExpression> columns = selectItems.getColumnExpression();
		Set<String> referencedFields = new HashSet<>();
		for (String alias : columns.keySet()) {
			ValueExpression expression = columns.get(alias);
			Set<String> fields = expression.collectVariableNames();
			for (String f : fields) {
				if(!usage.isDatasourceField(f)){
					throw new CPSException(ErrorCode.INPUT_ERROR, "Field " + f + " undefined in data sources");
				}
				if(usage.isOverlappingField(f)){
					Set<String> relatedDatasources = usage.getRelatedDatasources(f);
					throw new CPSException(ErrorCode.INPUT_ERROR, "Ambigious field " + f + " referenced. " + relatedDatasources + " are all having a field " + f + " defined, please specify the field by <Datasource>.<Field>");
				}
				String fullName = usage.getFullName(f);
				selectFieldsFullNames.add(fullName);
			}
			referencedFields.addAll(fields);
		}

		// Check if there is any select item alias that conflicts with data source fields
		for (String alias : columns.keySet()) {
			if(selectFieldsFullNames.contains(alias))
				throw new CPSException(ErrorCode.INPUT_ERROR, "Ambigious alias: " + alias + ", conflicting with existing datasource field");
			if(referencedFields.contains(alias))
				throw new CPSException(ErrorCode.INPUT_ERROR, "Ambigious alias: " + alias + ", conflicting with select item referenced field");
			usage.addAlias(alias);
			usage.addFullNameMapping(alias, alias);
			usage.addShortNameMapping(alias, alias);
		}

		// Update all fields in expression, converting them to full qualified name
		for (String alias : columns.keySet()) {
			ValueExpression expression = columns.get(alias);
			Set<String> names = expression.collectVariableNames();
			for (String shortName : names) {
				String fullName = usage.getFullName(shortName);
				expression.updateVariableName(shortName, fullName);
			}
		}

		// Collect all referenced field names
		for (String referencedField : referencedFields) {
			String fullName = usage.getFullName(referencedField);
			Set<String> fieldset =  usage.getRelatedDatasources(referencedField);
			Iterator<String> iterator = fieldset.iterator();
			String datasource = iterator.next();
			String shortName = usage.getShortName(fullName);
			usage.addFieldToProjection(datasource, shortName);
		}
	}

	private void collectUsageFromCondition(FieldUsage usage, BooleanExpression filterCondition) throws CPSException{
		if(filterCondition == null)
			return;
		Set<String> fields = filterCondition.collectVariableNames();
		for (String f : fields) {
			if(!usage.isAlias(f) && usage.isOverlappingField(f)){
				throw new CPSException(ErrorCode.INPUT_ERROR, "Ambigious field " + f + " used in filter conditions");
			}if(!usage.isDatasourceField(f)){
				throw new CPSException(ErrorCode.INPUT_ERROR, "Undefined field " + f + " used in filter conditions");
			}
			//Alias will not be projected at first projection
			if(!usage.isAlias(f)){
				String fullName = usage.getFullName(f);
				filterCondition.updateVariableName(f, fullName);
				String datasource = usage.getRelatedDatasources(fullName).iterator().next();
				String shortName = usage.getShortName(fullName);
				usage.addFieldToProjection(datasource, shortName);
			}
		}
	}

	private void updateAggregations(FieldUsage usage, AggregationDescriptor aggregation) throws CPSException{
		Set<String> keyFields = aggregation.getAggregationKeyFields();
		for (String field : keyFields) {
			if(!usage.isAlias(field) && usage.isOverlappingField(field))
				throw new CPSException(ErrorCode.INPUT_ERROR, "Ambigious field " + field + " used in aggregation");
			if(!usage.isDatasourceField(field))
				throw new CPSException(ErrorCode.INPUT_ERROR, "Undefined field" + field + " used in aggregation");
			String fullName = usage.getFullName(field);
			aggregation.updateAggregationKeyFields(field, fullName);
			String datasource = usage.getRelatedDatasources(field).iterator().next();
			String shortName = usage.getShortName(fullName);
			usage.addFieldToProjection(datasource, shortName);
		}
	}

	private void updateFilters(FieldUsage usage, BooleanExpression filter) throws CPSException{
		Set<String> fields = filter.collectVariableNames();
		for (String f : fields) {
			if(usage.isAlias(f))
				continue;
			if(usage.isOverlappingField(f))
				throw new CPSException(ErrorCode.INPUT_ERROR, "Ambigious field " + f + " used in filter conditions");
			if(!usage.isDatasourceField(f))
				throw new CPSException(ErrorCode.INPUT_ERROR, "Undefined field" + f + " used in filter conditions");
			String fullName = usage.getFullName(f);
			filter.updateVariableName(f, fullName);
			Set<String> datasourceset = usage.getRelatedDatasources(f);
			Iterator<String> iterator = datasourceset.iterator();
			String datasource = iterator.next();
//			String datasource = usage.getRelatedDatasources(f).iterator().next();
			String shortName = usage.getShortName(fullName);
			usage.addFieldToProjection(datasource, shortName);
		}
	}

	private static class FieldUsage{
		private Map<String, String> fullNameMapping = new HashMap<>();
		private Set<String> overlappingFields = new HashSet<>();
		private HashMultimap<String, String> fieldBelonging = HashMultimap.create();
		private Set<String> names = new HashSet<>();
		private Set<String> aliasSet = new HashSet<>();
		private Set<DatasourceField> fieldsForProjection = new HashSet<>();
		private Map<String, String> fullNameToShortName = new HashMap<>();

		public void addFieldToProjection(String datasourceName, String fieldName){
			fieldsForProjection.add(new DatasourceField(datasourceName, fieldName));
		}

		public Set<DatasourceField> getFieldsForProjection() {
			return fieldsForProjection;
		}

		public boolean isDatasourceField(String field){
			return names.contains(field);
		}

		public boolean isAlias(String alias){
			return aliasSet.contains(alias);
		}

		public boolean isOverlappingField(String field){
			return overlappingFields.contains(field);
		}

		public Set<String> getRelatedDatasources(String field){
			return fieldBelonging.get(field);
		}

		public String getFullName(String field){
			return fullNameMapping.get(field);
		}

		public void addAlias(String alias){
			aliasSet.add(alias);
		}

		public void addFieldBelonging(String field, String datasource){
			if(fieldBelonging.containsKey(field)){
				overlappingFields.add(field);
			}else{
				names.add(field);
			}
			fieldBelonging.put(field, datasource);
		}

		public void addFullNameMapping(String field, String fullName){
			fullNameMapping.put(field, fullName);
			names.add(field);
			names.add(fullName);
		}

		public void addShortNameMapping(String fullName, String shortName){
			fullNameToShortName.put(fullName, shortName);
			names.add(fullName);
			names.add(shortName);
		}

		public String getShortName(String fullName){
			return fullNameToShortName.get(fullName);
		}

	}

	public static class DatasourceField{
		private final String datasourceName;
		private final String fieldName;
		private final String fullQualifiedName;

		public DatasourceField(String datasourceName, String fieldName) {
			this.datasourceName = datasourceName;
			this.fieldName = fieldName;
			this.fullQualifiedName = SqlUtil.fullName(datasourceName, fieldName);
		}

		public String getDatasourceName() {
			return datasourceName;
		}

		public String getFieldName() {
			return fieldName;
		}

		public String fullQualifiedName(){
			return fullQualifiedName;
		}

		@Override
		public String toString() {
			return "{\"DatasourceField\":{\"datasourceName\":\""
					+ datasourceName + "\", \"fieldName\":\"" + fieldName
					+ "\"}}";
		}

		@Override
		public int hashCode() {
			final int prime = 31;
			int result = 1;
			result = prime
					* result
					+ ((datasourceName == null) ? 0 : datasourceName.hashCode());
			result = prime * result
					+ ((fieldName == null) ? 0 : fieldName.hashCode());
			return result;
		}

		@Override
		public boolean equals(Object obj) {
			if (this == obj)
				return true;
			if (obj == null)
				return false;
			if (getClass() != obj.getClass())
				return false;
			DatasourceField other = (DatasourceField) obj;
			if (datasourceName == null) {
				if (other.datasourceName != null)
					return false;
			} else if (!datasourceName.equals(other.datasourceName))
				return false;
			if (fieldName == null) {
				if (other.fieldName != null)
					return false;
			} else if (!fieldName.equals(other.fieldName))
				return false;
			return true;
		}

	}
}
